package com.sunday.common.mq.rabbit.study.base.e8_PublisherConfirms;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.sunday.common.mq.rabbit.study.base.ConnectionUtil;
import lombok.SneakyThrows;
import org.junit.jupiter.api.Test;

import java.time.Duration;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.function.BooleanSupplier;

/**
 * 发布者确认
 * <p>
 * 发布者确认是 RabbitMQ 扩展以实现可靠的 出版。当频道上启用了发布商确认时， 客户端发布的消息是异步确认的 由代理处理，这意味着它们已在服务器上得到处理 边。
 * <p>
 * 消息确认（publish confirms），设置管道为confirmSelect模式（channel.confirmSelect();）
 * 生产者与broker之间的消息确认称为public confirms，public confirms机制用于解决生产者与Rabbitmq服务器之间消息可靠传输，它在消息服务器持久化消息后通知消息生产者发送成功。
 * <p>
 * 总结
 * 确保已发布的消息到达代理在某些应用程序中可能是必不可少的。 发布者确认是 RabbitMQ 功能，有助于满足此要求。发行人 确认本质上是异步的，但也可以同步处理它们。 没有明确的方法来实现发布者确认，这通常会下降 到应用程序和整个系统中的约束。典型的技术有：
 * <p>
 * 单独发布消息，同步等待确认：简单，但非常 吞吐量有限。
 * 批量发布消息，同步等待批量确认：简单、合理 吞吐量，但很难推理何时出现问题。
 * 异步处理：最佳性能和资源利用率，在发生错误时具有良好的控制，但 可以参与正确实施。
 */
public class PublisherConfirms {

    static final int MESSAGE_COUNT = 50_000;

    /**
     * 策略#1：单独发布消息
     * 让我们从最简单的发布方法开始确认， 也就是说，发布消息并同步等待其确认：
     * 在前面的示例中，我们像往常一样发布一条消息并等待其 使用 Channel#waitForConconfirm OrDie（long） 方法进行确认。
     * 确认消息后，该方法将立即返回。如果 消息未在超时内确认，或者是否为裸消息（含义 由于某种原因，经纪人无法处理它），该方法将 引发异常。
     * 异常的处理通常包括 记录错误消息和/或重试发送消息。
     * <p>
     * 发布服务器是否确认异步？
     * 我们在开头提到经纪人确认已发布 消息异步，但在第一个示例中，代码等待 同步，直到消息被确认。客户实际上 异步接收确认，
     * 并相应地取消阻止等待确认或死亡的调用。将 waitForConfirmOrDie 视为同步助手 它依赖于后台的异步通知。
     */
    @Test
    @SneakyThrows
    public void publishMessagesIndividually() {

        ConnectionFactory cf = ConnectionUtil.factory();
        try (Connection connection = cf.newConnection()) {
            Channel ch = connection.createChannel();

            String queue = UUID.randomUUID().toString();
            ch.queueDeclare(queue, false, false, true, null);
            // 在此通道上启用发布者确认。
            ch.confirmSelect();
            long start = System.nanoTime();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String body = String.valueOf(i);
                ch.basicPublish("", queue, null, body.getBytes());
                /**
                 * 等待，直到自上次调用以来发布的所有消息都已完成
                 * 不是被经纪人骗了，就是被经纪人骗了;或者直到超时结束。
                 * 如果超时过期，则抛出TimeoutException。如果有
                 * 消息被删除，waitForConfirmsOrDie将抛出一个
                 * IOException。当在非确认频道上调用时，它将
                 * 抛出一个IllegalStateException。
                 */
                ch.waitForConfirmsOrDie(5_000);
            }
            long end = System.nanoTime();
            System.out.format("Published %,d messages individually in %,d ms%n", MESSAGE_COUNT, Duration.ofNanos(end - start).toMillis());
        }
    }

    /**
     * 策略#2：批量发布消息
     * 为了改进前面的示例，我们可以发布一个批处理 的消息，然后等待整个批次得到确认。 以下示例使用一批 100：
     * <p>
     * 等待一批消息被确认可显著提高吞吐量 等待单个消息的确认（使用远程 RabbitMQ 节点最多 20-30 次）。
     * 一个缺点是，如果发生故障，我们不知道到底出了什么问题， 因此，我们可能必须在内存中保留整个批处理以记录有意义的内容或 以重新发布消息。
     * 而且这个解决方案仍然是同步的，所以它 阻止发布消息。
     */
    @Test
    @SneakyThrows
    public void publishMessagesInBatch() {
        ConnectionFactory cf = ConnectionUtil.factory();
        try (Connection connection = cf.newConnection()) {
            Channel ch = connection.createChannel();

            String queue = UUID.randomUUID().toString();
            ch.queueDeclare(queue, false, false, true, null);

            ch.confirmSelect();

            int batchSize = 100;
            int outstandingMessageCount = 0;

            long start = System.nanoTime();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String body = String.valueOf(i);
                ch.basicPublish("", queue, null, body.getBytes());
                outstandingMessageCount++;

                if (outstandingMessageCount == batchSize) {
                    ch.waitForConfirmsOrDie(5_000);
                    outstandingMessageCount = 0;
                }
            }

            if (outstandingMessageCount > 0) {
                ch.waitForConfirmsOrDie(5_000);
            }
            long end = System.nanoTime();
            System.out.format("Published %,d messages in batch in %,d ms%n", MESSAGE_COUNT, Duration.ofNanos(end - start).toMillis());
        }
    }

    /**
     * 策略 #3：处理发布服务器异步确认
     * 代理异步确认已发布的消息，只需要 要在客户端上注册回调以通知这些确认：
     * <p>
     * 有 2 个回调：一个用于确认消息，一个用于裸消息 （可以认为代理丢失的消息）。每个回调都有 2 参数：
     * 序列号：标识已确认的数字 或裸色消息。我们将很快看到如何将其与已发布的消息相关联。
     * 倍数：这是一个布尔值。如果为 false，则只有一条消息被确认/裸读，如果 True，则所有序列号较小或相等的消息都将被确认/拒绝编辑。
     */
    @Test
    @SneakyThrows
    public void handlePublishConfirmsAsynchronously() {
        ConnectionFactory cf = ConnectionUtil.factory();
        try (Connection connection = cf.newConnection()) {
            Channel ch = connection.createChannel();

            String queue = UUID.randomUUID().toString();
            ch.queueDeclare(queue, false, false, true, null);

            ch.confirmSelect();

            ConcurrentNavigableMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();

            ConfirmCallback cleanOutstandingConfirms = (sequenceNumber, multiple) -> {
                if (multiple) {
                    ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(
                            sequenceNumber, true
                    );
                    confirmed.clear();
                } else {
                    outstandingConfirms.remove(sequenceNumber);
                }
            };

            // 代理异步确认已发布的消息，只需要 要在客户端上注册回调以通知这些确认
            ch.addConfirmListener(cleanOutstandingConfirms, (sequenceNumber, multiple) -> {
                // 发布代码现在使用地图跟踪出站消息。我们需要 在确认到达时清理此地图并执行诸如记录警告之类的操作 当消息为裸编辑时：
                String body = outstandingConfirms.get(sequenceNumber);
                System.err.format(
                        "Message with body %s has been nack-ed. Sequence number: %d, multiple: %b%n",
                        body, sequenceNumber, multiple
                );
                cleanOutstandingConfirms.handle(sequenceNumber, multiple);
            });

            long start = System.nanoTime();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String body = String.valueOf(i);
                // 序列号可以在发布前使用 Channel#getNextPublishSeqNo（） 获取：
                outstandingConfirms.put(ch.getNextPublishSeqNo(), body);
                ch.basicPublish("", queue, null, body.getBytes());
            }

            if (!waitUntil(Duration.ofSeconds(60), () -> outstandingConfirms.isEmpty())) {
                throw new IllegalStateException("All messages could not be confirmed in 60 seconds");
            }

            long end = System.nanoTime();
            System.out.format("Published %,d messages and handled confirms asynchronously in %,d ms%n", MESSAGE_COUNT, Duration.ofNanos(end - start).toMillis());
        }
    }

    static boolean waitUntil(Duration timeout, BooleanSupplier condition) throws InterruptedException {
        int waited = 0;
        while (!condition.getAsBoolean() && waited < timeout.toMillis()) {
            Thread.sleep(100L);
            waited += 100;
        }
        return condition.getAsBoolean();
    }

}
